package com.gugee.tiktok.data.consumer.keyword.listener;

import com.gugee.tiktok.data.common.msg.KeywordMessage;
import com.gugee.tiktok.data.consumer.keyword.config.KeywordAppConfig;
import com.gugee.tiktok.data.consumer.keyword.core.main.SaveAwemeKeywordMain;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 *
 * @author luolifeng
 * Date  2020/10/27
 */
@Component

public class KeywordMessageListener {

    @Autowired
    SaveAwemeKeywordMain saveAwemeKeywordMain;
    @Autowired
    KeywordAppConfig keywordAppConfig;
    protected final Logger log = LoggerFactory.getLogger(KeywordMessageListener.class);

    @KafkaListener(id="${kafka.consumer.group.keyword.id}",topics = {"${kafka.topics.keyword.online}"},containerFactory = "factoryKeywordAck")
    public void listenKeywordMessage(List<KeywordMessage> messages, Acknowledgment ack) {
        long start=System.currentTimeMillis();
        long acceptMsgTimeSecRange=keywordAppConfig.getAcceptMsgTimeMinRange()*60*1000;
        String error="success";
        try {
//            log.info("接收到关键词消息数量: {}" , messages.size());
            long isRange=start-acceptMsgTimeSecRange;
            /**
             * 消息如果不在处理时限内，则丢弃
             */
            if(messages.get(0).getEventTime()>isRange) {
                 saveAwemeKeywordMain.handle(messages);
            }else{
                log.info("关键词消息数量: {} ,不在处理时限内，跳过;第一消息时间：{},时限分钟：{}" , messages.size(),messages.get(0).getEventTime(),keywordAppConfig.getAcceptMsgTimeMinRange() );
            }
            ack.acknowledge();
        }catch (Exception e){
            log.error("发生异常：{}" ,e.getMessage());
            error=e.getMessage();
        }finally {
            long times=(System.currentTimeMillis()-start)/1000;
            if(times<1){
                times=1;
            }
            log.info("关键词消息数量: {},处理耗时: {}s,每秒处理消息数={}/s,结果：{}" , messages.size(),times,messages.size()/times,error);
        }
    }

}
